#include "ConnPool.h"

pthread_mutex_t ConnPool::_mutex_init = PTHREAD_MUTEX_INITIALIZER;
ConnPool *ConnPool::_instance = NULL;

ConnPool *ConnPool::Instance()
{
    if(_instance == NULL)
    {
        pthread_mutex_lock(&_mutex_init);

        if(_instance == NULL)
        {
            int32 socketPoolSize   = ConfManager::Instance()->GetSocketPollSize();
            int32 socketAccpetSize = ConfManager::Instance()->GetSocketAcceptSize();
            int32 epollTimeOut     = ConfManager::Instance()->GetEpollTimeOut();
            int32 port             = ConfManager::Instance()->GetListenPort();
            _instance = new ConnPool(port, socketPoolSize, socketAccpetSize, epollTimeOut);

            WriteLog(NOTICE, "Create ConnPool port[%d] socketPoolSize[%d] socketAccpetSize[%d] epollTimeOut[%d] successfully.",
                    port, socketPoolSize, socketAccpetSize, epollTimeOut);
        }

        pthread_mutex_unlock(&_mutex_init);
    }

    return _instance;
}

ConnPool::ConnPool()
{
    _epollFd          = -1;
    _listenFd         = -1;
    _port             = -1;
    _isRunning        = false;
    _socketPoolSize   = -1;
    _socketAccpetSize = -1;
    _epollTimeOut     = -1;
    _pEvents          = NULL;

    pthread_mutex_init(&_mutex, NULL);
    pthread_cond_init(&_notEmptyCond, NULL);
}

ConnPool::ConnPool(int32 port, int32 socketPollSize, int32 socketAcceptSize, int32 epollTimeOut)
{
    _epollFd       = -1;
    _listenFd      = -1;
    _port          = -1;
    _isRunning     = false;

    if(port <= 0)
    {
        WriteLog(WARNING, "Init ConnPool error, listen port[%d] must bigger then 0", port);
        exit(-1);
    }
    _port = port;

    if(socketPollSize <= 0)
    {
        WriteLog(WARNING, "Init ConnPool error, socketPollSize[%d] must bigger then 0", socketPollSize);
        exit(-1);
    }
    _socketPoolSize = socketPollSize;

    if(socketAcceptSize <= 0)
    {
        WriteLog(WARNING, "Init ConnPool error, socketAcceptSize[%d] must bigger then 0", socketAcceptSize);
        exit(-1);
    }
    _socketAccpetSize = socketAcceptSize;

    if(epollTimeOut <= 0)
    {
        WriteLog(WARNING, "Init ConnPool error, epollTimeOut[%d] must bigger then 0", epollTimeOut);
        exit(-1);
    }
    _epollTimeOut = epollTimeOut;

    _pEvents      = NULL;

    pthread_mutex_init(&_mutex, NULL);
    pthread_cond_init(&_notEmptyCond, NULL);
}

ConnPool::~ConnPool()
{
    if(_pEvents != NULL)
    {
        delete [] _pEvents;
        _pEvents = NULL;
    }

    if(_listenFd > 0)
    {
        close(_listenFd);
        _listenFd = -1;
    }

    if(_epollFd > 0)
    {
        close(_epollFd);
        _epollFd = -1;
    }

    pthread_mutex_destroy(&_mutex);
    pthread_cond_destroy(&_notEmptyCond);
}

bool ConnPool::_EpollAdd(int32 fd)
{
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = EPOLLIN | EPOLLHUP | EPOLLERR | EPOLLONESHOT;
    if(epoll_ctl(_epollFd, EPOLL_CTL_ADD, fd, &ev) != 0)
    {
        WriteLog(WARNING, "EpollAdd error[%d] fd[%d]", errno, fd);
        return false;
    }
    return true;
}

bool ConnPool::_EpollMod(int32 fd)
{
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = EPOLLIN | EPOLLHUP | EPOLLERR | EPOLLONESHOT;
    if(epoll_ctl(_epollFd, EPOLL_CTL_MOD, fd, &ev) != 0)
    {
        WriteLog(WARNING, "EpollMod error[%d] fd[%d]", errno, fd);
        return false;
    }
    return true;
}

bool ConnPool::_EpollDel(int32 fd)
{
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = EPOLLIN | EPOLLHUP | EPOLLERR | EPOLLONESHOT;
    epoll_ctl(_epollFd, EPOLL_CTL_DEL, fd, &ev);

    return true;
}

void ConnPool::_CloseSocket(Socket socket)
{
    pthread_mutex_lock(&_mutex);

    close(socket.fd);
    std::map<int32, Socket>::iterator it = _socketInfoContainer.find(socket.fd);
    if(it != _socketInfoContainer.end())
    {
        if(it->second.isInEpoll)
        {
            _EpollDel(socket.fd);
        }
        _socketInfoContainer.erase(it);
    }

    pthread_mutex_unlock(&_mutex);
}

void ConnPool::_KeepSocket(Socket socket)
{
    pthread_mutex_lock(&_mutex);

    std::map<int32, Socket>::iterator it = _socketInfoContainer.find(socket.fd);
    if(it != _socketInfoContainer.end())
    {
        if(it->second.isInEpoll == false)
        {
            if(!_EpollAdd(socket.fd))
            {
                close(socket.fd);
                _socketInfoContainer.erase(it);
            }
            else
            {
                it->second.isInEpoll = true;
            }
        }
        else
        {
            if(!_EpollMod(socket.fd))
            {
                close(socket.fd);
                _EpollDel(socket.fd);
                _socketInfoContainer.erase(it);
            }
        }
    }
    else
    {
        close(socket.fd);
    }

    pthread_mutex_unlock(&_mutex);
}

bool ConnPool::IsRunning()
{
    return _isRunning || ActiveSocketSize();
}

void ConnPool::Start()
{
    WriteLog(NOTICE, "ConnPool is starting......");

    if(_epollFd >= 0)
    {
        close(_epollFd);
        _epollFd = -1;
    }
    if((_epollFd = epoll_create(_socketPoolSize+1)) < 0)
    {
        WriteLog(FATAL, "Create epoll error[%d].", _epollFd);
        exit(-1);
    }

    if(_pEvents != NULL)
    {
        delete [] _pEvents;
        _pEvents = NULL;
    }
    if((_pEvents = new epoll_event[_socketPoolSize]) == NULL)
    {
        WriteLog(FATAL, "New epoll_event[%d] error.", _socketPoolSize);
        exit(-1);
    }

    if(_listenFd >= 0)
    {
        close(_listenFd);
        _listenFd = -1;
    }
    if((_listenFd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
    {
        WriteLog(FATAL, "Create socket error[%d].", _listenFd);
        exit(-1);
    }

    struct sockaddr_in* svrAddr = new sockaddr_in();
    memset(svrAddr, 0, sizeof(struct sockaddr_in));
    svrAddr->sin_family = AF_INET;
    svrAddr->sin_addr.s_addr = htonl(INADDR_ANY);
    svrAddr->sin_port = htons(_port);

    int ret;
    if((ret = bind(_listenFd, (struct sockaddr*)svrAddr, sizeof(struct sockaddr_in))) < 0)
    {
        WriteLog(FATAL, "Bind socket error[%d].", ret);
        exit(-1);
    }

    if((ret = listen(_listenFd, _socketAccpetSize)) < 0)
    {
        WriteLog(FATAL, "Listen socket error[%d].", ret);
        exit(-1);
    }

    struct epoll_event event;
    event.data.fd = _listenFd;
    event.events  = EPOLLIN|EPOLLET;
    if((ret = epoll_ctl(_epollFd, EPOLL_CTL_ADD, _listenFd, &event)) < 0)
    {
        WriteLog(FATAL, "Add listenFd to epoll error[%d].", ret);
        exit(-1);
    }

    _isRunning = true;

    WriteLog(NOTICE, "ConnPool start successfully, listening at %d, socket pool size[%d].", _port, _socketPoolSize);
}

void ConnPool::Stop()
{
    WriteLog(NOTICE, "ConnPool is stopping......");

    _isRunning = false;

    if(_listenFd >=0)
    {
        close(_listenFd);
        _listenFd = -1;
    }

    if(ActiveSocketSize() != 0)
    {
        sleep(2);
    }

    pthread_mutex_lock(&_mutex);

    WriteLog(NOTICE, "There are %u client in ConnPoll.", (uint32)_socketInfoContainer.size());
    std::map<int32, Socket>::iterator it;
    for(it=_socketInfoContainer.begin(); it!=_socketInfoContainer.end(); it++)
    {
        close(it->second.fd);
        WriteLog(NOTICE, "Client fd[%d] close successfully.", it->second.fd);
    }

    pthread_mutex_unlock(&_mutex);

    WriteLog(NOTICE, "ConnPool stop successfully.");
}

bool ConnPool::CheckSocket()
{
    int eTotal = epoll_wait(_epollFd, _pEvents, _socketPoolSize, _epollTimeOut);
    if(eTotal <= 0)
    {
        return false;
    }
    sockaddr_t clientAddr;
    Socket socket;
    for(int i=0; i<eTotal; i++)
    {
        if (_pEvents[i].data.fd == _listenFd)
        {
            memset(&clientAddr, 0, sizeof(sockaddr_t));
            uint32 addrLen = sizeof(sockaddr_t);
            int clientFd = 0;
            if((clientFd = accept(_listenFd, &clientAddr, &addrLen)) < 0)
            {
                continue;
            }
            socket.fd = clientFd;
            socket.ip = GetIP((struct sockaddr_in*)&clientAddr);
            socket.port = GetPort((struct sockaddr_in*)&clientAddr);
            socket.isInEpoll = false;
            _PushSocket(socket);
        }
        else if(_pEvents[i].events & EPOLLIN)
        {
            socket.fd = _pEvents[i].data.fd;
            _PushSocket(socket);
        }
        else
        {
            socket.fd = _pEvents[i].data.fd;
            WriteLog(WARNING, "epoll_wait error, fd[%d].", socket.fd);
            _CloseSocket(socket);
        }
    }

    return true;
}

bool ConnPool::_PushSocket(Socket socket)
{
    pthread_mutex_lock(&_mutex);

    std::map<int32, Socket>::iterator it = _socketInfoContainer.find(socket.fd);
    if(it != _socketInfoContainer.end())
    {
        it->second.isInEpoll = true;
        _socketContainer.push(socket.fd);
        pthread_cond_signal(&_notEmptyCond);
    }
    else if( _socketInfoContainer.size() < (uint32)_socketPoolSize)
    {
        if(_EpollAdd(socket.fd))
        {
            socket.isInEpoll = true;
            _socketInfoContainer.insert(std::map<int32, Socket>::value_type(socket.fd, socket));
        }
        else
        {
            close(socket.fd);
        }
    }
    else
    {
        WriteLog(WARNING, "Socket Container is full[%d], refuse the new socket[%d].", (int32)_socketInfoContainer.size(), (int32)(socket.fd));
        close(socket.fd);
    }

    pthread_mutex_unlock(&_mutex);
    return true;
}

bool ConnPool::PopSocket(Socket &socket)
{
    pthread_mutex_lock(&_mutex);

    struct timespec ts;
    int32 errNo = 0;
    bool ret = false;
    while (_isRunning && _socketContainer.size()<=0)
    {
        ts.tv_sec = time(NULL) + 2;
        ts.tv_nsec = 0;
        errNo = pthread_cond_timedwait(&_notEmptyCond, &_mutex, &ts);
        if(errNo == EPERM)
        {
            WriteLog(WARNING, "Socket Pool is empty, Pop socket pthread_cond_timedwait time out and get mutex error.");
            pthread_mutex_lock(&_mutex);
        }
    }

    if(_socketContainer.size()>0)
    {
        int32 fd = _socketContainer.front();
        _socketContainer.pop();
        std::map<int32, Socket>::iterator it = _socketInfoContainer.find(fd);
        if(it != _socketInfoContainer.end())
        {
            socket = it->second;
            ret = true;
        }
    }

    pthread_mutex_unlock(&_mutex);
    return ret;
}

void ConnPool::ResetSocket(Socket socket, int32 keepAlive)
{
    if(keepAlive == 0)
    {
        _CloseSocket(socket);
    }
    else
    {
        _KeepSocket(socket);
    }
}

int32 ConnPool::ActiveSocketSize()
{
    pthread_mutex_lock(&_mutex);

    int32 size = (int32)_socketContainer.size();

    pthread_mutex_unlock(&_mutex);
    return size;
}

int32 ConnPool::EpollSocketSize()
{
    pthread_mutex_lock(&_mutex);

    int32 size = (int32)_socketInfoContainer.size();

    pthread_mutex_unlock(&_mutex);
    return size;
}
